package net.bwie.realtime.jtp.dwd.log.function1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction extends ProcessFunction<String,String> {
    private final OutputTag<String> errorLog;
    private final OutputTag<String> startLog;
    private final OutputTag<String> displayLog;
    private final OutputTag<String> actionLog;
    public AppLogSplitProcessFunction(OutputTag<String> errorLog, OutputTag<String> startLog,
                                      OutputTag<String> displayLog, OutputTag<String> actionLog){
        this.errorLog=errorLog;
        this.startLog=startLog;
        this.displayLog=displayLog;
        this.actionLog=actionLog;
    }
    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        //错误日志
        if(jsonObject.getJSONObject("err") != null){
            ctx.output(errorLog,jsonObject.toJSONString());
            jsonObject.remove("err");
        }
        //启动日志
        if(jsonObject.getJSONObject("start") != null){
            ctx.output(startLog,jsonObject.toJSONString());
        }
        else {
            //获取曝光和动作
            JSONArray displayArray = jsonObject.getJSONArray("display");
            JSONArray actionsArray = jsonObject.getJSONArray("actions");
            jsonObject.remove("display");
            jsonObject.remove("actions");
            out.collect(jsonObject.toJSONString());
            if(displayArray != null && displayArray.isEmpty()){
                for(int i = 0;i < displayArray.size();i++){
                    JSONArray displayJsonObject = displayArray.getJSONArray(i);
                    jsonObject.put("display",displayJsonObject);
                    ctx.output(displayLog,jsonObject.toJSONString());
                }
            }
            if(actionsArray != null && actionsArray.isEmpty()){
                for(int i = 0;i < actionsArray.size();i++){
                    JSONArray actionsJsonObject = actionsArray.getJSONArray(i);
                    jsonObject.put("actions",actionsJsonObject);
                    ctx.output(actionLog,jsonObject.toJSONString());
                }
            }
        }
    }
}
